package cn.azzhu.day04.async;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.elasticsearch.action.ActionFuture;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.transport.client.PreBuiltTransportClient;

import java.net.InetAddress;
import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import java.util.function.Supplier;

/**
 * @author azzhu
 * @create 2020-09-19 16:39:10
 */
public class AsyncQueryFromES extends RichAsyncFunction<String, Tuple2<String,String>> {
    private  static transient TransportClient transportClient;
    @Override
    public void open(Configuration parameters) throws Exception {
        //设置集群名称
        final Settings settings = Settings.builder().put("cluster.name", "docker-cluster").build();
        //创建Client
        transportClient = new PreBuiltTransportClient(settings).addTransportAddress(
                new InetSocketTransportAddress(InetAddress.getByName("localhost"), 9300));
    }

    @Override
    public void close() throws Exception {
        transportClient.close();
    }

    @Override
    public void asyncInvoke(String key, ResultFuture<Tuple2<String, String>> resultFuture) throws Exception {
        final ActionFuture<GetResponse> actionFuture = transportClient.get(new GetRequest("azzhu", "users", key));

        CompletableFuture.supplyAsync(new Supplier<String>() {
            @Override
            public String get() {
                try {
                    final GetResponse response = actionFuture.get();
                    return response.getSource().get("name").toString();
                } catch (Exception e) {
                    e.printStackTrace();
                    return null ;
                }
            }
        }).thenAccept((String dbResult) -> {
            resultFuture.complete(Collections.singleton(new Tuple2<>(key,dbResult)));
        });
    }
}
